package com.wyx.springbootspringkafkademo.consumer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.wyx.springbootspringkafkademo.entity.Message;
import com.wyx.springbootspringkafkademo.producer.KafkaProducerService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

/**
 * @author wyx
 * @Date 2025-03-26
 */
@Slf4j
@Component
public class KafkaConsumerService {

  /**
   * 单消息消费，手动提交offset
   * @param messageStr
   * @param acknowledgment
   */
  @KafkaListener(topics = KafkaProducerService.TOPIC)
  public void consume(@Payload String messageStr, Acknowledgment acknowledgment) {
    log.info("Received messageStr: {}", messageStr);
    Message message = null;
    try {
      ObjectMapper mapper = new ObjectMapper();
      message = mapper.readValue(messageStr, Message.class);
      System.out.println("Received Message: " + message);
    } catch (JsonProcessingException e) {
      System.err.println("Error parsing message: " + e.getMessage());
    }

    if (message.getId() % 2 == 0) {
      //      throw new RuntimeException("id error");
    }
    // 提交offset
    acknowledgment.acknowledge();
  }
}
